package com.atguigu.state;

/*
* @date 2024/11/22 8:53
*
* 对账案例需求：
*   我们可以实现一个实时对账的需求，也就是app的支付操作和第三方的支付操作的一个双流Join
*   App的支付事件和第三方的支付事件会互相等待5秒钟，如果等不来对应的支付事件，那么就输出报警信息
*
* 思路：主要用来练习定时器的使用
*    1.创建两条流：app支付流和第三方支付流
*    2.过滤掉其它数据，只保留支付数据
*    3.将两条流connect到一起，并使用keyBy,相同key放到一起
*    4.分别对两条流中数据进行处理，设置定时器
*    5.设置两个中间状态，分别用来保存App的支付事件的状态和第三方的支付事件的状态
*
* */
//todo 写完了 已测
//多条流时，下游收到水位线以最小的为准，触发下游定时器的是最小的水位线
//todo 水位线在数据后面，所以当相应先算数据后更新水位线
import com.atguigu.pojo.AppEvent;
import com.atguigu.pojo.ThirdPartyEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink11_BillCheck {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 1.分别创建两条流：支付流
        // AppEvent: 1001,order-1,pay,1000
        SingleOutputStreamOperator<AppEvent> appDs = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new AppEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), Long.valueOf(fields[3].trim()));
                        }
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<AppEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );


        //todo 1.分别创建两条流：第三方支付流
        // ThirdPartyEvent: 1001,order-1,pay,wechat,2000
        SingleOutputStreamOperator<ThirdPartyEvent> thirdPartyDs = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new ThirdPartyEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), fields[3].trim() , Long.valueOf(fields[4].trim()));
                        }
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ThirdPartyEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );
        thirdPartyDs.print("TPD");

        //过滤appDs的支付数据
        SingleOutputStreamOperator<AppEvent> appFilterDs = appDs.filter(
                event -> "pay".equals(event.getEventType())
        );

        appFilterDs.print("APP");

        //TODO 将两条流连接到一起connect，相同key放一起，并使用定时器功能（必须keyBy）
        SingleOutputStreamOperator<String> processDs = appFilterDs
                .keyBy(AppEvent::getOrderId)
                .connect(thirdPartyDs.keyBy(ThirdPartyEvent::getOrderId))
                .process(
                        new KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>() {

                            //1.声明状态
                            private ValueState<AppEvent> appState;

                            private ValueState<ThirdPartyEvent> thirdPartyState;

                            //2.open初始化状态

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //创建上下文对象，通过上下文对象来获取当前状态赋值给声明的状态
                                RuntimeContext runtimeContext = getRuntimeContext();

                                ValueStateDescriptor<AppEvent> appEventStateDesc = new ValueStateDescriptor<>("appEventStateDesc", Types.POJO(AppEvent.class));
                                appState = runtimeContext.getState(appEventStateDesc);

                                ValueStateDescriptor<ThirdPartyEvent> thirdPartyEventStateDesc = new ValueStateDescriptor<>("thirdPartyEventStateDesc", Types.POJO(ThirdPartyEvent.class));
                                thirdPartyState = runtimeContext.getState(thirdPartyEventStateDesc);
                            }

                            /**
                             * 1.处理appDs数据流
                             *
                             * 1.1 AppEvent到了，到状态中找ThirdPartyEvent
                             *
                             * 1.2如果找到，说明ThirdPartyEvent先到，从状态中取出ThirdPartyEvent,对账成功，删除ThirdPartyEvent定时器，清除状态
                             *
                             * 1.3如果找不到，说明AppEvent先到，将AppEvent放到状态中，并注册5秒后的定时器
                             *
                             * 1.4如果定时器触发了，对账失败
                             *
                             */

                            @Override
                            public void processElement1(AppEvent appEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> out) throws Exception {
                                //1.AppEvent到了，到状态中找ThirdPartyEvent
                                ThirdPartyEvent thirdPartyEvent = thirdPartyState.value();

                                if (thirdPartyEvent != null) {
                                    //2. 如果找到， 说明ThirdPartyEvent先到， 从状态中取出ThirdPartyEvent, 对账成功， 删除ThirdPartyEvent定时器 , 清除状态
                                    //输出对账成功
                                    out.collect(appEvent.getOrderId() + "对账成功，ThirdPartyEvent先到，AppEvent后到");
                                    //删除定时器
                                    context.timerService().deleteEventTimeTimer(thirdPartyEvent.getTs() + 5000L);
                                    //清除状态
                                    thirdPartyState.clear();
                                } else {
                                    appState.update(appEvent);
                                    context.timerService().registerEventTimeTimer(appEvent.getTs() + 5000L);
                                }
                            }

                            /**
                             * 处理thirdPartyDs
                             *
                             * 1. ThirdPartyEvent 到了 ， 到状态中找 AppEvent
                             * 2. 如果找到， 说明 AppEvent 先到， 从状态中取出 AppEvent, 对账成功， 删除 AppEvent 定时器, 清除状态
                             * 3. 如果找不到， 说明 ThirdPartyEvent 先到， 将 ThirdPartyEvent 放到状态中， 并注册5秒后的定时器
                             * 4. 如果定时器触发了， 对账失败。
                             */

                            @Override
                            public void processElement2(ThirdPartyEvent thirdPartyEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> out) throws Exception {
                                // 1. ThirdPartyEvent 到了 ， 到状态中找 AppEvent
                                AppEvent appEvent = appState.value();


                                if (appEvent != null) {
                                    //2. 如果找到， 说明 AppEvent 先到， 从状态中取出 AppEvent, 对账成功， 删除 AppEvent 定时器, 清除状态
                                    //对账成功
                                    out.collect(thirdPartyEvent.getOrderId() + "对账成功， AppEvent先到，ThirdPartyEvent后到");


                                    // 删除定时器
                                    context.timerService().deleteEventTimeTimer(appEvent.getTs() + 5000L);
                                    // 清除状态
                                    appState.clear();
                                    // 3. 如果找不到， 说明 ThirdPartyEvent 先到， 将 ThirdPartyEvent 放到状态中， 并注册5秒后的定时器
                                } else {
                                    thirdPartyState.update(thirdPartyEvent);
                                    context.timerService().registerEventTimeTimer(thirdPartyEvent.getTs() + 5000L);
                                }

                            }

                            @Override
                            public void onTimer(long timestamp, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                                // 4. 如果定时器触发了， 对账失败。
                                if (appState.value() != null) {
                                    out.collect(ctx.getCurrentKey() + " 对账失败, AppEvent到了, ThirdPartyEvent未到!!!!");
                                } else if (thirdPartyState.value() != null) {
                                    out.collect(ctx.getCurrentKey() + " 对账失败, ThirdPartyEvent到了, AppEvent未到!!!!");
                                }
                                //清除状态
                                appState.clear();
                                thirdPartyState.clear();
                            }
                        }
                );

        processDs.print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
